package com.dtwave.cfstreaming.sink.function;

import com.dtwave.cfstreaming.bean.ResultData;
import com.dtwave.cfstreaming.connection.ElasticConnector;
import com.dtwave.cfstreaming.utils.ElasticUtils;
import com.dtwave.param.EsparamObj;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RestHighLevelClient;


/**
 * 构造并返回我们需要的IndexRequest
 */
public class EsSinkFunction implements ElasticsearchSinkFunction<ResultData> {
    RestHighLevelClient client;

    private EsparamObj esparamObj;
    public EsSinkFunction(EsparamObj esparamObj) {
        this.esparamObj = esparamObj;
    }

    @Override
    public void open() throws Exception {
        client = ElasticConnector.getClient("es连接地址");
    }


    @Override
    public void process(ResultData resultData, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
        IndexRequest indexRequest = ElasticUtils.createIndexAndQuery(client, esparamObj, resultData);
        if(indexRequest!=null){
            requestIndexer.add(indexRequest);
        }
    }


    @Override
    public void close() throws Exception {
        if(client != null){
            client.close();
        }
    }

}
